package com.hugedata.idc.etl;

import java.io.DataInputStream;
import java.io.IOException;
import java.util.Calendar;

import org.apache.commons.compress.archivers.ArchiveException;
import org.apache.commons.compress.archivers.ArchiveInputStream;
import org.apache.commons.compress.archivers.ArchiveStreamFactory;
import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.log4j.Logger;

import com.hugedata.idc.utils.HDFSUtils;

public class MergeHDFS
{
    private static final Logger LOGGER = Logger.getLogger(MergeHDFS.class);

    public static void main(String[] args) throws IOException, ClassNotFoundException
    {
        String hdfsTempDir = args[0];
        String hdfsMergeDir = args[1];
        String types = args[2];
        boolean isDel = Boolean.getBoolean(args[3]);
        
        LOGGER.info("hdfsTempDir " + hdfsTempDir);
        LOGGER.info("hdfsMergeDir " + hdfsMergeDir);
        LOGGER.info("types " + types);
        LOGGER.info("isDel " + isDel);
        
//        String hdfsTempDir = "/idc/qostest1/test1";
//        String hdfsMergeDir = "/idc/qostest1/testmerge1";
//        String types = "qos,traceroute";
//        boolean isDel = false;
        mergeByType(hdfsTempDir, hdfsMergeDir, types, isDel);
    }

    /**
     * 根据类型合并文件
     * 
     * @param hdfsTempDir
     * @param hdfsMergeDir
     * @param types
     *            crawl,qos,dnsdelay
     * @param isDel
     * @throws IOException
     * @throws ClassNotFoundException
     */
    public static void mergeByType(String hdfsTempDir, String hdfsMergeDir, String types, boolean isDel) throws IOException,
            ClassNotFoundException
    {
        Configuration config = HDFSUtils.getConf();
        FileSystem fs = FileSystem.get(config);
        Path srcPath = new Path(hdfsTempDir);
        String[] splits = types.split(",");
        for (String type : splits)
        {
            PathFilter filter = new LogFilter("_" + type + "_");
            FileStatus[] listStatus = fs.listStatus(srcPath, filter);
            String mergeFileName = DateFormatUtils.format(Calendar.getInstance(), "yyyyMMdd") + "_" + type;
            Path destPath = new Path(hdfsMergeDir + "/" + type, mergeFileName);
            for (FileStatus fileStatus : listStatus)
            {
                Path path = fileStatus.getPath();
                String tempFileName = path.getName();
                LOGGER.info("tar name-----" + tempFileName);
                DataInputStream inputStream = fs.open(path);
                ArchiveInputStream stream;
                try
                {
                    if (StringUtils.contains(tempFileName, ".tar.gz"))
                    {
                        stream = new ArchiveStreamFactory().createArchiveInputStream("tar", new GzipCompressorInputStream(inputStream));
                    }
                    else
                    {
                        stream = new ArchiveStreamFactory().createArchiveInputStream("tar", new BZip2CompressorInputStream(
                                inputStream));
                    }
                }
                catch (ArchiveException e)
                {
                    LOGGER.error("文件解压失败");
                    throw new IOException(e);
                }
                FSDataOutputStream outputStream = null;
                try
                {
                    if (fs.exists(destPath))
                    {
                        outputStream = fs.append(destPath);
                    }
                    else
                    {
                        outputStream = fs.create(destPath);
                    }
                    while (stream.getNextEntry() != null)
                    {
                        IOUtils.copy(stream, outputStream);
                    }
                }
                finally
                {
                    IOUtils.closeQuietly(inputStream);
                    IOUtils.closeQuietly(outputStream);
                }
                if (isDel)
                {
                    fs.delete(path, true);
                }
            }
        }

    }

}
